﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Threading;
using ons;

namespace test
{
 
    public class MyMsgListener : MessageListener
    {
        public MyMsgListener()
        {
        }

        ~MyMsgListener()
        {
        }

        public override ons.Action consume(Message value, ConsumeContext context)
        {
            try
            {
                Byte[] text = Encoding.UTF8.GetBytes(value.getBody());
                Console.WriteLine("Receive message:{0}", value.getMsgID());
                Console.WriteLine(Encoding.UTF8.GetString(text));
            }
            catch(Exception ex)
            {
                //Byte[] text = Encoding.UTF8.GetBytes(value.getBody());
                //Console.WriteLine("Receive message:{0}", value.getMsgID());
                //Console.WriteLine(Encoding.UTF8.GetString(text));
            }


            //Byte[] text = Encoding.Default.GetBytes(value.getBody());
            //Console.WriteLine("Receive message:{0}", value.getMsgID());
            //Console.WriteLine(Encoding.UTF8.GetString(text));


            return ons.Action.CommitMessage;
        }
    }


    public class MyMsgOrderListener : MessageOrderListener
    {
        public MyMsgOrderListener()
        {

        }

        ~MyMsgOrderListener()
        {
        }

        public override ons.OrderAction consume(Message value, ConsumeOrderContext context)
        {
            Byte[] text = Encoding.Default.GetBytes(value.getBody());
            Console.WriteLine(Encoding.UTF8.GetString(text));
            return ons.OrderAction.Success;
        }
    }


    public class OnscSharp
    {
        private static Producer _producer;
        private static PushConsumer _consumer;
        private static OrderConsumer _orderconsumer;
        private static OrderProducer _orderproducer;
        private static MyMsgListener _listen;
        private static MyMsgOrderListener _order_listen;
        private static object s_SyncLock = new Object();
        private static string Ons_Topic = "device_data";
        private static string Ons_AccessKey = "LTAI4FmVyq7hNNjtxRJJVxex";
        private static string Ons_SecretKey = "02RF6wIc1uqy7VPsXiUvu5yie5M6hU";
        private static string Ons_GroupId = "GID_Meter1";
        private static string Ons_NameSrv = "http://MQ_INST_1701930854581949_BbxI5hS0.mq-internet-access.mq-internet.aliyuncs.com:80";

        public static void SendMessage(string msgBody, String tag = "RegisterLog")
        {
            Message msg = new Message(Ons_Topic, tag, msgBody);
            msg.setKey(Guid.NewGuid().ToString());
            try
            {
                SendResultONS sendResult = _producer.send(msg);
                Console.WriteLine("send success {0}", sendResult.getMessageId());
            }
            catch (Exception ex)
            {
                Console.WriteLine("send failure{0}", ex.ToString());
            }
        }

        public static void SendOrderMessage(string msgBody, String tag = "RegisterLog", String key = "test")
        {
            Message msg = new Message(Ons_Topic, tag, msgBody);
            msg.setKey(Guid.NewGuid().ToString());
            try
            {
                SendResultONS sendResult = _orderproducer.send(msg, key);
                Console.WriteLine("send success {0}", sendResult.getMessageId());
            }
            catch (Exception ex)
            {
                Console.WriteLine("send failure{0}", ex.ToString());
            }
        }

        public static void StartPushConsumer()
        {
            _listen = new MyMsgListener();
            _consumer.subscribe(Ons_Topic, "*", _listen);
            _consumer.start();
        }

        public static void StartOrderConsumer()
        {
            _order_listen = new MyMsgOrderListener();
            _orderconsumer.subscribe(Ons_Topic, "*", _order_listen);
            _orderconsumer.start();
        }

        public static void shutdownPushConsumer()
        {
            _consumer.shutdown();
        }

        public static void shutdownOrderConsumer()
        {
            _orderconsumer.shutdown();
        }

        public static void StartProducer()
        {
            _producer.start();
        }

        public static void ShutdownProducer()
        {
            _producer.shutdown();
        }


        public static void StartOrderProducer()
        {
            _orderproducer.start();
        }

        public static void ShutdownOrderProducer()
        {
            _orderproducer.shutdown();
        }

        private static ONSFactoryProperty getFactoryProperty()
        {
            ONSFactoryProperty factoryInfo = new ONSFactoryProperty();
            factoryInfo.setFactoryProperty(ONSFactoryProperty.AccessKey, Ons_AccessKey);
            factoryInfo.setFactoryProperty(ONSFactoryProperty.SecretKey, Ons_SecretKey);
            factoryInfo.setFactoryProperty(ONSFactoryProperty.ConsumerId, Ons_GroupId);
            factoryInfo.setFactoryProperty(ONSFactoryProperty.ProducerId, Ons_GroupId);
            factoryInfo.setFactoryProperty(ONSFactoryProperty.PublishTopics, Ons_Topic);
            factoryInfo.setFactoryProperty(ONSFactoryProperty.NAMESRV_ADDR, Ons_NameSrv);
            factoryInfo.setFactoryProperty(ONSFactoryProperty.LogPath, "C://log");
           return factoryInfo;
        }

        public static void  CreatePushConsumer()
        {

            _consumer = ONSFactory.getInstance().createPushConsumer(getFactoryProperty());
        }

        public static void CreateProducer()
        {

            _producer = ONSFactory.getInstance().createProducer(getFactoryProperty());
        }


        public static void CreateOrderConsumer()
        {

            _orderconsumer = ONSFactory.getInstance().createOrderConsumer(getFactoryProperty());
        }

        public static void CreateOrderProducer()
        {

            _orderproducer = ONSFactory.getInstance().createOrderProducer(getFactoryProperty());
        }
    }

    class ConsumerAndProducerForEx
    {
        static void Main(string[] args)
        {
             OnscSharp.CreateProducer();
             OnscSharp.CreatePushConsumer();
             OnscSharp.StartPushConsumer();
             OnscSharp.StartProducer();
             System.DateTime beforDt = System.DateTime.Now;
            // for (int i = 0;i < 10; ++i)
            // {
            //    //byte[] bytes = Encoding.UTF8.GetBytes("中文messages");//中文encode
            //    //String body = Convert.ToBase64String(bytes);
            //    OnscSharp.SendMessage("This is test message");
            //    Thread.Sleep(1000 * 1);
            //}
             System.DateTime endDt = System.DateTime.Now;
             System.TimeSpan ts = endDt.Subtract(beforDt);
             Console.WriteLine("per message:{0}ms.", ts.TotalMilliseconds / 10000);
             Thread.Sleep(1000 * 100);
             Console.ReadKey();
             OnscSharp.ShutdownProducer();
             OnscSharp.shutdownPushConsumer();
             Console.WriteLine("end");
        }
    }
}
